package com.shen.jcbf.jdk8bf.forkjoin;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;

/**
 * 说明：jdk7 的 ForkJoin 测试2，实例理解
 * 
 * @author shenhx
 * @date 2017年6月20日
 */
public class ForkJoinTest2 {

	/**
	 * 说明：
	 * ForkJoinTask ：是一个将在ForkJoinPool中执行的任务的基类
	 * Fork/Join 框架提供了在一个任务里执行fork和join操作的机制和控制任务状态的方法，通常，为了实现Fork/Join任务，
	 * 需要实现以下两个类之一的子类。
	 * 
	 * RecursiveAction ：用于任务没有返回结果的场景,一个ForkJoinTask任务类，递归无结果的任务类。类似callable一样的线程任务，
	 * RecursiveTask ： 用于任务有返回结果的场景
	 * 
	 * @author shenhx
	 * @date 2017年6月20日
	 * @param args
	 * @throws InterruptedException
	 */
	public static void main(String[] args) throws InterruptedException {
		// 生成商品数据
		List<Product> list = new ArrayList<Product>();
		for (int i = 0; i < 40; i++) {
			Product p = new Product("苹果" + i, 10);
			list.add(p);
		}
		// 这个类实现了 ExecutorServic接口和工作窃取算法。它管理工作者线程，并提供任务的状态信息，以及任务的执行信息
		ForkJoinPool fjp = new ForkJoinPool();

		Task task = new Task(list, 0, list.size(), 19);
		
		fjp.execute(task);
		
//		fjp.shutdown(); //关闭线程池
//		fjp.awaitTermination(1, TimeUnit.MINUTES);
		//等待超时。结合shutdown来让任务一完成就继续执行下面的代码

		//System.out.println("---------");
		
		// 使用循环的方式来查看任务的信息
		do {
			//Thread.sleep(5000);
			System.out.printf("活跃线程：%s,这一个参数  %s,并行执行的最大数量：%s\n", fjp.getActiveThreadCount(), fjp.getStealCount(),fjp.getParallelism());
		} while (!task.isDone()); // 如果任务还未完成，则继续获取信息
		
		// 如果这个任务完成没有抛出异常并没有取消。
		if (task.isCompletedNormally()) {
			System.out.println("main:任务完成");
		}else{
			System.out.println("任务失败。。");
		}
		
		System.out.println("main:------------------------------  打印任务结果");
		for (Product product : list) {
			int price = product.getPrice();
			String name = product.getName();
			if (price != 19) { // 结果不是所期望的。就打印出来
				System.out.println(name + "," + price);
			}
		}
		System.out.println("main:------------------------------  打印任务结束");
		
	}
}

class Product {
	private String name;
	private int price;

	public Product(String name, int price) {
		this.name = name;
		this.price = price;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public int getPrice() {
		return price;
	}

	public void setPrice(int price) {
		this.price = price;
	}

}

class Task extends RecursiveAction {
	private List<Product> list; // 所有任务
	private int start; // 处理任务的开始索引
	private int end; // 处理任务的结束索引
	private int price; // 更改的价格

	public Task(List<Product> list, int start, int end, int price) {
		this.list = list;
		this.start = start;
		this.end = end;
		this.price = price;
	}

	@Override
	protected void compute() {
		//任务异常调试。
		//int i = 1/0;
		if (end - start <= 10) { // 每个task 只能处理10条数据。
			System.out.printf("起始：start:%s,end:%s\n", start, end);
			update();
		} else { // 多余的数据，则需要分给更多的任务
			int middle = (end + start) / 2; // 因为是索引。所以需要开始和结尾相加，然后除以2 就能得到
											// 两个索引之间的数值
			Task task1 = new Task(list, start, middle, 19);
			Task task2 = new Task(list, middle, end, 19);

			System.out.printf("分析：middle:%s,start:%s,end:%s\n", middle, start, end); // 方便推算
			// 这里把任务分成了2半递归执行。
			invokeAll(task1, task2);
		}
	}

	// 根据给定的起始索引和结束索引更新结果
	private void update() {
		for (int i = start; i < end; i++) {
			Product product = list.get(i);
			product.setPrice(price);

			System.out.printf("%s,修改了价格,索引：%s,%s,%s\n", Thread.currentThread().getName(), i, product.getName(),
					product.getPrice());
		}
	}
}
